package demo.hxzon.mapreduce;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import com.google.common.base.Function;
import com.google.common.collect.ImmutableListMultimap;
import com.google.common.collect.Multimaps;

public class MapReduce {

    public static final void main(String[] args) {
        Employee guest = new Builder("Guest").build();
        Employee anna = new Builder("Anna").build();
        Employee thomas = new Builder("Thomas").age(41).build();
        Employee luke = new Builder("Luke").company("LucasArt").build();
        Employee yoda = new Builder("Yoda").age(800).company("LucasArt").build();

        Collection<Employee> employees = new ArrayList<Employee>();
        employees.add(guest);
        employees.add(anna);
        employees.add(thomas);
        employees.add(luke);
        employees.add(yoda);

        ImmutableListMultimap<String, Employee> personsGroupByCompany = Multimaps.index(employees, new Function<Employee, String>() {

            public String apply(Employee person) {
                return person.getCompany();
            }

        });
        System.out.println(personsGroupByCompany);
        //{DevCode=[Employee [name=Guest, age=30,company=DevCode], 
        //Employee [name=Anna, age=30,company=DevCode], Employee [name=Thomas, age=41,company=DevCode]], 
        //LucasArt=[Employee [name=Luke, age=30,company=LucasArt], Employee [name=Yoda, age=800,company=LucasArt]]}

        List<Employee> averageAgeByCompany = new ArrayList<Employee>();

        for (String company : personsGroupByCompany.keySet()) {
            List<Employee> employeesForThisCompany = personsGroupByCompany.get(company);
            int sum = 0;
            for (Employee employee : employeesForThisCompany) {
                sum += employee.getAge();
            }
            averageAgeByCompany.add(new Employee("average", sum / employeesForThisCompany.size(), company));
        }
        System.out.println("Result: " + averageAgeByCompany);
        //Result: [Employee [name=average, age=33,company=DevCode], Employee [name=average, age=415,company=LucasArt]]

    }
//scala
//case class Employee(name: String = "guest", age: Int = 30, company: String = "DevCode")
//
//object MapReduce {
//  def main(args: Array[String]): Unit = {
//
//    val guest = Employee()
//    val anna = Employee("Anna")
//    val thomas = Employee("Thomas", 41)
//    val luke = Employee("Luke", company = "LucasArt")
//    val yoda = luke.copy("Yoda", age = 800)
//
//    val allEmployees = List(luke, anna, guest, yoda, thomas)
//    val sortedEmployees = allEmployees.groupBy(_.company)
//    println(sortedEmployees)
//    val averageAgeByCompany = sortedEmployees.map {
//      case (key, value) =>
//        value(0).copy(name = "average", age = (value.map(_.age).sum) / value.size)
//    }
//    println("Result: " + averageAgeByCompany)
//  }
//}
}
